package org.wejob;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.function.Consumer;

public class ZkClient {
    public CuratorFramework client;
    private ZkClient() {}
    public ZkClient (String zkAddress) {
        if (client == null) {
            synchronized (ZkClient.class) {
                if (client == null) {
                    client = CuratorFrameworkFactory.builder().
                            connectString(zkAddress).
                            sessionTimeoutMs(5000).
                            connectionTimeoutMs(10000).
                            retryPolicy(new ZkRetryPolicy().retryPolicy3).
                            namespace("wejob").
                            build();
                }
            }
        }
    }

    public ZkClient start() {
        if (client != null && !client.getState().equals(CuratorFrameworkState.STARTED)) {
            client.start();
        }
        return this;
    }

    public ZkClient close() {
        if (client != null && !client.getState().equals(CuratorFrameworkState.STOPPED)) {
            client.close();
        }
        return this;
    }

    public void getLeaderAddress(Consumer<String> callback) throws Exception {
        //访问主节点，主节点不存在则抢占选举；主节点存在则获取主节点信息，进入同步模式
        Stat stat = client.checkExists().forPath("/leader");
        if (stat == null) {//节点不存在--尝试创建节点--这里判断主节点是否存在是在集群初始化的时候
            System.out.println("主节点不存在,尝试创建。。。");
            client.create().withMode(CreateMode.EPHEMERAL).forPath("/leader", "myAddress".getBytes());
            callback.accept("i am leader");
        } else {
            byte[] bytes = client.getData().forPath("/leader");
            callback.accept(new String(bytes));
        }
    }
}
